package com.open.source.demo;

import com.open.source.MqConstant;
import com.open.source.kafka.AbstractKafkaManualCommitConsumerListener;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

/**
 * @author ZonLen since on 2022/8/11 上午10:34
 */
@Slf4j
@Component
public class DemoKafkaManualCommitConsumerListener extends
    AbstractKafkaManualCommitConsumerListener {


  @Override
  public String kafkaTopic() {
    return MqConstant.TAX_TOPIC_DELAY_DEMO;
  }

  @Override
  public String deserializerEncoding() {
    return "GBK";
  }

//  @Override
//  public String groupId() {
//    return "ZHANG-LONG";
//  }

  @Override
  public void onMessage(ConsumerRecord<String, String> record, Acknowledgment acknowledgment) {
    log.info("Manual offset ＝{}, val={} ", record.offset(), record.value());
//    acknowledgment.nack(2000);
    acknowledgment.acknowledge();
//    int a = 1/0;
  }

}
